package com.bigdata.core.transformations

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.{ArrayBuffer, ListBuffer}

/**
 * mapPartitionsWithIndex
 * 可以拿到每个RDD中的分区，以及分区中的数据
 */
object Demo18_mapPartitionWithIndex {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("mapPartitionWithIndex")
    val sc = new SparkContext(conf)
    sc.setLogLevel("error")

    val lines: RDD[String] = sc.textFile("data/words", 3)

    val result: RDD[String] = lines.mapPartitionsWithIndex((index: Int, iter: Iterator[String]) => {
      val list = new ListBuffer[String]
      println(s"partition index $index")
      // 把一行的数据切分并返回
      iter.foreach(line => {
        println(s"line --- $line")
        val arr: Array[String] = line.split(" ")
        arr.foreach(
          list += _
        )
      })
      list.iterator
    })
    result.foreach(println)
  }
}
